package cas.ihep.hmss.queue;

import cas.ihep.hmss.Job;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.File;
import java.io.IOException;
import java.util.LinkedList;
import java.util.ListIterator;

/**
 * Created by chocolate on 2017/12/1.
 */
public class HmssQueue implements JobQueue {

    private final LinkedList<Job> hmsQue;
    private final FileSystem fs;

    public HmssQueue() throws IOException {
        hmsQue = new LinkedList<Job>();
        fs = FileSystem.get(new Configuration());
    }

    public void enqueue(Job job) {
        synchronized (hmsQue) {
            hmsQue.offerLast(job);
        }
    }

    public Job dequeue(int s) {
        synchronized (hmsQue) {
            if (hmsQue.isEmpty()) {
                return null;
            }
            if (hmsQue.size() == 1) {
                return hmsQue.removeFirst();
            } else {
                try {
                    ListIterator<Job> iterator = hmsQue.listIterator();
                    double priIndex = -1;
                    long minLen = Integer.MAX_VALUE;
                    long minTime = Integer.MAX_VALUE;
                    long maxLen = -1;
                    long maxTime = -1;
                    long current = System.currentTimeMillis();
                    while (iterator.hasNext()) {
                        Job job = iterator.next();
                        System.out.println(current-job.enqueueTime);
                        if (current - job.enqueueTime < minTime) {
                            minTime = current - job.enqueueTime;
                        }
                        if (current - job.enqueueTime > maxTime) {
                            maxTime = current - job.enqueueTime;
                        }
                        if (job.length < 0) {
                            job.length = 0;
                            if (job.destPath.contains("hdfs://")) {
                                for (int i = 0; i < job.srcPath.size(); i++) {
                                    String p = job.srcPath.get(i);
                                    File file = new File(p);
                                    job.length += file.length();
                                }
                            } else {
                                Path p;
                                for (int i = 0; i < job.srcPath.size(); i++) {
                                    p = new Path(job.srcPath.get(i));
                                    job.length += fs.getFileStatus(p).getLen();
                                }

                            }
                        }
                        if (job.length < minLen) {
                            minLen = job.length;
                        }
                        if (job.length > maxLen) {
                            maxLen = job.length;
                        }

                    }

                    Job subJob = null;
                    iterator = hmsQue.listIterator();
                    System.out.println("Min len: "+minLen);
                    System.out.println("Max len: "+maxLen);
                    while (iterator.hasNext()) {

                        Job j = iterator.next();
                        j.p1 = (double) (j.priority - 1) / (9 - 1);
                        j.p3 = ((double) j.length / s - 1.0 * minLen / s) / (1.0 * maxLen / s - 1.0 * minLen / s);
                        if (maxTime==minTime){
                            j.p2=1;
                        }else {
                            j.p2 = (double) ((current - j.enqueueTime) - minTime) / (maxTime - minTime);
                        }
                        j.jobIndex = 1.68 * j.p1 + 1.2 * j.p2 + 1 * (1 / (j.p3 / s + 1));
                        StringBuilder str = new StringBuilder();
                        str.append(j.p1).append(',').append(j.p2).append(',').append(j.p3).append(',').append(j.jobIndex);
                        System.out.println(str.toString());
                        //System.out.println(s);
                        //System.out.println(maxTime);
                        //System.out.println(minTime);
                        //System.out.println(current-j.enqueueTime);
                        //System.out.println(hmsQue.size());
                        //System.out.println(j.enqueueTime);
                        //System.out.println(j.srcPath.get(1));
                        if (j.jobIndex > priIndex) {
                            priIndex = j.jobIndex;
                            subJob = j;
                        } else if (j.jobIndex == priIndex) {
                            if (subJob == null || j.priority > subJob.priority) {
                                subJob = j;
                            }
                        }

                    }
                    hmsQue.remove(subJob);
                    return subJob;

                }catch(Exception ignored){

                    return null;
                }
            }
        }
    }
}